package com.gradle.quartz.batch.config;

import com.gradle.bean.employee.Employee;
import com.gradle.quartz.batch.partitioner.CustomerPartitioner;
import com.gradle.quartz.batch.service.EmployeeService;
import org.apache.ibatis.session.SqlSessionFactory;
import org.mybatis.spring.batch.MyBatisBatchItemWriter;
import org.mybatis.spring.batch.MyBatisPagingItemReader;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.core.partition.PartitionHandler;
import org.springframework.batch.core.partition.support.TaskExecutorPartitionHandler;
import org.springframework.batch.item.validator.BeanValidatingItemProcessor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.scheduling.annotation.Async;

import java.util.HashMap;
import java.util.Map;

@EnableBatchProcessing
@Configuration
@Async
public class PartitionerConfig {
    @Autowired
    private JobBuilderFactory jobBuilderFactory;
    @Autowired
    private StepBuilderFactory stepBuilderFactory;
    @Autowired
    private SqlSessionFactory sqlSessionFactory;
    @Autowired
    private EmployeeService employeeService;
    @Autowired
    private CustomerPartitioner customerPartitioner;
    /**
     * 从步骤读
     *
     * @param form  mapp.xml参数
     * @param to    mapp.xml参数
     * @param range 数据分段间隔
     * @return MyBatisPagingItemReader<Employee>
     */
    @Bean
    @StepScope
    public MyBatisPagingItemReader<Employee> myBatisPagingItemReader(
            @Value("#{stepExecutionContext[startIndex]}") Integer form,
            @Value("#{stepExecutionContext[toIndex]}") Integer to,
            @Value("#{stepExecutionContext[range]}") Integer range) {
        StringBuilder builder = new StringBuilder();
        builder.append("---MyBatisPagingItemReader开始---")
                .append("----from：").append(form)
                .append("----to：").append(to)
                .append("----每片数量：").append(range);
        System.out.println(builder.toString());
        MyBatisPagingItemReader<Employee> reader = new MyBatisPagingItemReader<>();
        reader.setSqlSessionFactory(sqlSessionFactory);
        //_pagesize和_skiprows由mybatis自动处理
        reader.setPageSize(1000);
        reader.setQueryId("com.gradle.mapper.EmployeeMapper.queryTempForList");
        Map<String, Object> map = new HashMap<>();
        map.put("startIndex", form);
        map.put("toIndex", to);
        // from to 两个参数在分区器中配置，从上下文传送
        reader.setParameterValues(map);
        return reader;
    }

    /**
     * 从步骤写
     *
     * @return MyBatisBatchItemWriter<Employee>
     */
    @Bean
    public MyBatisBatchItemWriter<Employee> dbToDbItemWriter() {
        MyBatisBatchItemWriter<Employee> writer = new MyBatisBatchItemWriter<>();
        //构建sqlSession类似Connection
        writer.setSqlSessionFactory(sqlSessionFactory);
        //指定需要执行sql语句对应mapper语句
        writer.setStatementId("com.gradle.mapper.EmployeeMapper.save");
        return writer;
    }

    @Bean
    public BeanValidatingItemProcessor<Employee> partitionerItemProcessor() {
        System.out.println("---------------执行参数校验------------------");
        BeanValidatingItemProcessor<Employee> processor = new BeanValidatingItemProcessor<>();
        //不满足条件丢弃
        processor.setFilter(true);
        return processor;
    }

    /**
     * 从步骤
     *
     * @return
     */
    @Bean
    public Step dbToDbWorkStep() {
        System.out.println(">>>>>>>>>分区步骤的逻辑对象【step】");
        return stepBuilderFactory.get("dbToDbWorkStep")
                //一次性读多少数据,以及数据类型为User
                .<Employee, Employee>chunk(1000)
                .reader(myBatisPagingItemReader(null, null, null))
                .writer(dbToDbItemWriter())
                .processor(partitionerItemProcessor())
                .build();
    }

    /**
     * 分区处理器
     *
     * @return PartitionHandler
     */
    @Bean
    public PartitionHandler dbToDbHandler() {
        TaskExecutorPartitionHandler handler = new TaskExecutorPartitionHandler();
        int count = employeeService.countEmployeeTemp();
        //从步骤个数
        handler.setGridSize(count / 10000);
        //一个从步骤就是一个独立的线程
        handler.setTaskExecutor(new SimpleAsyncTaskExecutor());
        //关联从步骤
        handler.setStep(dbToDbWorkStep());
        try {
            handler.afterPropertiesSet();
        } catch (Exception exception) {
            System.out.println(exception.getMessage());
        }
        return handler;
    }

    /**
     * 主步骤
     *
     * @return Step
     */
    @Bean
    public Step masterStep() {
        return stepBuilderFactory.get("masterStep")
                .partitioner(dbToDbWorkStep().getName(), customerPartitioner)
                .partitionHandler(dbToDbHandler())
                .build();
    }

    /**
     * 作业job
     *
     * @return Job
     * @throws Exception
     */
    @Bean
    public Job dbToDbjob() throws Exception {
        System.out.println(">>>>>>>>>主步骤流程【dbToDbjob】>>>>>>>>");
        return jobBuilderFactory.get("partitioner-step-job")
                .start(masterStep())
                .incrementer(new RunIdIncrementer())
                .build();
    }
}
